package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

//用户域用户注册事实表
public class DwdUserRegister {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_user_register_group"));
        Table userInfo = tableEnv.sqlQuery("select " +
                " data['id'],\n" +
                " data['login_name'],\n" +
                " data['nick_name'],\n" +
                " data['passwd'],\n" +
                " data['real_name'],\n" +
                " data['phone_num'],\n" +
                " data['email'],\n" +
                " data['head_img'],\n" +
                " data['user_level'],\n" +
                " data['birthday'],\n" +
                " data['gender'],\n" +
                " data['create_time'],\n" +
                " data['operate_time'],\n" +
                " data['status']," +
                " ts" +
                " from topic_db " +
                " where `table` = 'user_info' and `type` = 'insert'");
        tableEnv.createTemporaryView("user_info",userInfo);
        tableEnv.executeSql("create table dwd_user_register( " +
                " id string,\n" +
                " login_name string,\n" +
                " nick_name string,\n" +
                " passwd string,\n" +
                " real_name string,\n" +
                " phone_num string,\n" +
                " email string,\n" +
                " head_img string,\n" +
                " user_level string,\n" +
                " birthday string,\n" +
                " gender string,\n" +
                " create_time string,\n" +
                " operate_time string,\n" +
                " status string," +
                " ts string," +
                " primary key(id) not enforced" +
                " ) " + KafkaUtil.getUpsertKafkaDDL("dwd_user_register"));

        tableEnv.executeSql("insert into dwd_user_register select * from user_info");
    }
}
